package com.andyyan.elasticsearch.client.impl;

import com.andyyan.elasticsearch.client.def.model.ESClient;
import com.andyyan.elasticsearch.client.def.model.Highlight;
import com.andyyan.elasticsearch.client.def.model.HighlightItem;
import com.andyyan.elasticsearch.client.def.model.Item;
import com.andyyan.elasticsearch.client.def.query.ESQueryRequest;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Joiner;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsResponse;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by yantingxin on 16-9-8.
 */
public class ESClientImpl implements ESClient, InitializingBean {

    Logger logger = LoggerFactory.getLogger(ESClientImpl.class);

    private ObjectMapper mapper = new ObjectMapper();

    private final String startTag = "|";

    private final String endTag = "|";

    private ESClientPool clientPool;

    //创建对象的map
    @Override
    public boolean addMapping(String index, String type, String mapping) {
        Client client = null;
        try {
            client = getClient();
            CreateIndexResponse response = client.admin().indices().prepareCreate(index).addMapping(type, mapping).get();
            return true;
        } catch (Exception e) {
            logger.error("ESClientImpl addMapping error", e);
        } finally {
            releaseClient(client);
        }
        return false;
    }

    @Override
    public boolean delIndex(String index) {
        Client client = null;
        try {
            client = getClient();
            if (client.admin().indices().prepareExists(index).get().isExists()) {
                client.admin().indices().prepareDelete(index).get();
                return true;
            }
        } catch (Exception e) {
            logger.error("ESClientImpl delIndex error", e);
        } finally {
            releaseClient(client);
        }
        return false;
    }

    //保存单个对象
    @Override
    public boolean addDocument(String index, String type, String id, Object object) {
        Client client = null;
        try {
            client = clientPool.getClient();
            byte[] json = mapper.writeValueAsBytes(object);
            client.prepareIndex(index, type, id).setSource(json).get();
            return true;
        } catch (Exception e) {
            logger.error("ESClientImpl addDocument error", e);
        } finally {
            releaseClient(client);
        }
        return false;
    }

    @Override
    public String getDocumentJson(String index, String type, String id) {
        Client client = null;
        try {
            client = clientPool.getClient();
            GetRequest getRequest = new GetRequest(index, type, id);
            GetResponse response = client.get(getRequest).actionGet();
            String value = response.getSourceAsString();
            return value;
        } catch (Exception e) {
            logger.error("ESClientImpl getDocumentJson error", e);
        } finally {
            releaseClient(client);
        }
        return null;
    }

    @Override
    public <T> T getDocument(String index, String type, String id, Class<T> clazz) {
        Client client = null;
        try {
            client = clientPool.getClient();
            GetRequest getRequest = new GetRequest(index, type, id);
            GetResponse response = client.get(getRequest).actionGet();
            T t = clazz.newInstance();
            BeanUtils.populate(t, response.getSource());
            return t;
        } catch (Exception e) {
            logger.error("ESClientImpl getDocument error", e);
        } finally {
            releaseClient(client);
        }
        return null;
    }

    //删除
    @Override
    public boolean delDocument(String index, String type, String id) {
        Client client = null;
        try {
            client = getClient();
            client.prepareDelete(index, type, id).get();
            return true;
        } catch (Exception e) {
            logger.error("ESClientImpl delDocument error", e);
        } finally {
            releaseClient(client);
        }
        return false;
    }

    //判断是否存在index
    @Override
    public boolean existIndex(String index) {
        Client client = null;

        try {
            if (StringUtils.isNotBlank(index)) {
                client = getClient();
                return client.admin().indices().prepareExists(index).get().isExists();
            }
        } catch (Exception e) {
            logger.error("ESClientImpl existIndex error", e);
        } finally {
            releaseClient(client);

        }
        return false;
    }

    //判断是否存在type
    @Override
    public boolean existType(String index, String type) {
        return existType(new String[]{index}, new String[]{type});
    }

    @Override
    public boolean existType(String[] indexes, String[] types) {
        Client client = null;

        try {
            if (indexes != null && indexes.length > 0 && types != null && types.length > 0) {
                client = getClient();
                TypesExistsRequest request = new TypesExistsRequest();
                request.indices(indexes);
                request.types(types);
                TypesExistsResponse response = client.admin().indices().typesExists(request).actionGet();
                return response.isExists();
//                return client.admin().indices().prepareTypesExists(type).get().isExists();
            }
        } catch (Exception e) {
            logger.error("ESClientImpl existType error", e);
        } finally {
            releaseClient(client);

        }
        return false;
    }

    @Override
    public String getMappingsJson(String index, String type) {
        Client client = null;

        try {
            if (index != null && type != null) {
                client = getClient();
                GetMappingsRequest request = new GetMappingsRequest();
                request.indices(index);
                request.types(type);
                GetMappingsResponse response = client.admin().indices().getMappings(request).actionGet();
                ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings = response.getMappings();



                return mappings.toString();
            }
        } catch (Exception e) {
            logger.error("ESClientImpl getMapping error", e);
        } finally {
            releaseClient(client);

        }
        return "";
    }

    //批量保存对象
    @Override
    public boolean batchAddDocument(String index, String type, Map<String, Object> objectMap) {
        Client client = null;

        try {
            client = getClient();
            // 批量处理request
            BulkRequestBuilder bulkRequest = client.prepareBulk();

            byte[] json;
            for (String key : objectMap.keySet()) {
                try {
                    json = mapper.writeValueAsBytes(objectMap.get(key));
                    bulkRequest.add(new IndexRequest(index, type, key).source(json));
                } catch (IOException e) {
                    logger.error("ESClientImpl batchAddDocument for error", e);
                }
            }

            // 执行批量处理request
            BulkResponse bulkResponse = bulkRequest.get();
            boolean result = !bulkResponse.hasFailures();
            // 处理错误信息
            dealFailLog(bulkResponse, "创建");

            return result;
        } catch (Exception e) {
            logger.error("ESClientImpl batchAddDocument error", e);
        } finally {
            releaseClient(client);
        }
        return false;
    }

    @Override
    public boolean batchDelDocument(String index, String type, List<String> idList) {
        Client client = null;
        try {
            if (CollectionUtils.isNotEmpty(idList)) {
                client = getClient();
                BulkRequestBuilder requestBuilder = client.prepareBulk();
                for (String id : idList) {
                    requestBuilder.add(new DeleteRequest(index, type, id));
                }
                BulkResponse bulkResponse = requestBuilder.get();
                boolean result = !bulkResponse.hasFailures();
                // 处理错误信息
                dealFailLog(bulkResponse, "删除");

                return result;
            }
        } catch (Exception e) {
            logger.error("ESClientImpl deleteDocument error", e);
        } finally {
            releaseClient(client);
        }
        return false;
    }

    //查询列表
    @Override
    public <T> List<Item<T>> query(ESQueryRequest request, Class<T> clazz) {
        Client client = null;
        List<Item<T>> itemList = null;
        try {
            client = getClient();
            SearchRequestBuilder requestBuilder = client.prepareSearch(request.getIndexName())
                    .setTypes(request.getTypeName())
//                    .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                    .setSearchType(request.getSearchType());
            if (request.getOffset() >= 0) {
                requestBuilder.setFrom(request.getOffset());
            }
            if (request.getLimit() > 0) {
                requestBuilder.setSize(request.getLimit());
            }

            if (CollectionUtils.isNotEmpty(request.getHighlightFieldList())) {
                for (String field : request.getHighlightFieldList()) {
                    requestBuilder.addHighlightedField(field);
                }

                requestBuilder.setHighlighterPreTags("|")
                        .setHighlighterPostTags("|");
            }

            requestBuilder.setQuery(request.getQueryBuilder());


            SearchResponse response = requestBuilder.execute().actionGet(request.getTimeout());



            SearchHit[] hits = response.getHits().getHits();

            if (hits != null && hits.length > 0) {
                itemList = new ArrayList<Item<T>>(request.getLimit());
            }

            for (SearchHit hit : hits) {
                Item<T> item = new Item<T>();
                T value = clazz.newInstance();
                BeanUtils.populate(value, hit.getSource());
                item.setObject(value);

                //处理多个字段高亮逻辑
                if (CollectionUtils.isNotEmpty(request.getHighlightFieldList())) {
                    for (String field : request.getHighlightFieldList()) {
                        Object fieldValue = hit.getSource().get(field);
                        if (fieldValue != null) {
                            List<Highlight> highlightList = getHighlightList(hit, field);
                            HighlightItem searchItem = new HighlightItem();
                            searchItem.setField(field);
                            searchItem.setContent(fieldValue.toString());
                            searchItem.setHighlightList(highlightList);
                            item.putHighlightItem(field, searchItem);
                        }
                    }
                }

                itemList.add(item);
            }

        } catch (Exception e) {
            e.printStackTrace();
            logger.error("ESClientImpl search error", e);
        } finally {
            releaseClient(client);
        }
        return itemList;
    }

    private void dealFailLog(BulkResponse responses, String action) {
        try {
            // 处理错误信息
            if (responses.hasFailures()) {
                String error = "====================批量" + action +"索引过程中出现错误 下面是错误信息==========================\n";
                long count = 0L;
                for (BulkItemResponse bulkItemResponse : responses) {
                    error += ("发生错误的 索引id为 : "+bulkItemResponse.getId()+" ，错误信息为："+ bulkItemResponse.getFailureMessage() + "\n");
                    count++;
                }
                error += ("====================批量" + action + "索引过程中出现错误 上面是错误信息 共有: "+count+" 条记录==========================\n");

                logger.error("ESClientImpl bulkRequest error:" + error);
            }
        } catch (Exception e) {
            logger.error("ESClientImpl dealFailLog error", e);
        }
    }

    private List<Highlight> getHighlightList(SearchHit hit, String field) {
        List<Highlight> highlightList = new ArrayList<Highlight>();
        if (hit.getHighlightFields() != null && hit.getHighlightFields().size() > 0) {
            String highlightText = Joiner.on("").join(hit.getHighlightFields().get(field).getFragments()).replace(startTag + endTag, "");
            int start = -1, end = -1;
            int count = 0;
            for (int i = 0; i < highlightText.length(); i++) {
                if (highlightText.charAt(i) == '|' && start == -1) {
                    start = i - count * 2;
                } else if (highlightText.charAt(i) == '|' && end == -1) {
                    end = i - (count + 1) * 2 + 1;
                    if (start > -1 && end > start) {
                        highlightList.add(new Highlight(start, end));
                        start = -1;
                        end = -1;
                    }
                    count++;
                }
            }
        }
        return highlightList;
    }

    private Client getClient() {
        return clientPool.getClient();
    }

    private void releaseClient(Client client) {
        clientPool.releaseClient(client);
    }

    public void setClientPool(ESClientPool clientPool) {
        this.clientPool = clientPool;
    }

    public void afterPropertiesSet() throws Exception {
        Assert.notNull(clientPool, "clientPool is required");
    }
}
